package com.icbc.dccsh.bolt;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import com.icbc.dccsh.util.Watcher;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class NeedHistoryBolt extends BaseBasicBolt {
	private Watcher _watcher = null;
	@Override
    public void prepare(Map stormConf, TopologyContext context) {
		this._watcher = Watcher.getWatcher(this.getClass());
    }
	public void execute(Tuple input, BasicOutputCollector collector) {
		this._watcher.count();
		HashMap<String, Object> input_hashmap = (HashMap<String, Object>)input.getValueByField("input");
		HashMap<String, Object> mapping_hashmap = (HashMap<String, Object>)input.getValueByField("mapping");
		
		boolean needHistoryData = false;
		for (Map.Entry<String, Object> entry : mapping_hashmap.entrySet()){
			String metricName = entry.getKey();
			HashMap<String, Object> metricMapping = (HashMap<String, Object>)entry.getValue();
			String metricFormula = (String)metricMapping.get("formula");
			if (metricFormula.indexOf("___prior") > 0 || metricFormula.indexOf("delta") > 0){
				needHistoryData = true;
			}
		}
		
		
//		ArrayList metricList = (ArrayList)mapping_hashmap.get("metrics");
//		for (Object metricMapping_o : metricList){ // for each metric
//			HashMap<String, Object> metricMapping = (HashMap<String, Object>)metricMapping_o;
//			String metricFormula = (String)metricMapping.get("formula"); // get metric formula
//			if (metricFormula.indexOf("___prior") > 0 || metricFormula.indexOf("delta") > 0){
//				needHistoryData = true;
//			}
//		}
		
		if (needHistoryData){
			collector.emit("get_history_stream", new Values(input_hashmap, mapping_hashmap));
		}
		else {
			collector.emit("no_history_stream", new Values(input_hashmap, mapping_hashmap));
		}
	}
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declareStream("get_history_stream", new Fields("input", "mapping"));
		declarer.declareStream("no_history_stream", new Fields("input", "mapping"));
	}
}
